package insert;

import java.awt.List;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Random;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import conf.UserConfigure;

import util.MetaDataHandler;

public class insertTableDynamic {
	private static Configuration configuration = UserConfigure.configuration;

	/**
	 * @param args
	 */
	public static void main(String[] args) {
		Configuration conf = null;
		conf = HBaseConfiguration.create();
		conf.addResource(new Path("./src/conf/user-connect.xml"));
//			configuration.set("hbase.master", "192.9.200.172:60000");
//			configuration.set("hbase.zookeeper.quorum","192.9.200.10");  
//			configuration.set("hbase.zookeeper.property.clientPort", "2181");    
//			configuration.set("hbase.HServerAddress", "192.9.200.172:60000");
		// TODO Auto-generated method stub

		// conf.addResource("hbase-site-cluster.xml");//可以指定文件加载
//		HBaseAdmin hAdmin = new HBaseAdmin(conf);
		/**========创建表=========*/
//		HTableDescriptor desc = new HTableDescriptor("tableDynamic");
//		desc.addFamily(new HColumnDescriptor("cf1"));
//		desc.addFamily(new HColumnDescriptor("cf2"));
//		desc.addFamily(new HColumnDescriptor("cf3"));
//		desc.addFamily(new HColumnDescriptor("cf4"));
//		desc.addFamily(new HColumnDescriptor("cf5"));
//		hAdmin.createTable(desc );

		/**=========插入数据=========*/
//		Put put = new Put(Bytes.toBytes("1"));
//		put.add(Bytes.toBytes("article"), Bytes.toBytes("title"), Bytes.toBytes("Head First HBase"));
//		put.add(Bytes.toBytes("article"), Bytes.toBytes("content"), Bytes.toBytes("HBase is the Hadoop database. Use it when you need random, realtime read/write access to your Big Data."));
//		put.add(Bytes.toBytes("article"), Bytes.toBytes("tags"), Bytes.toBytes("Hadoop,HBase,NoSQL"));
//		put.add(Bytes.toBytes("author"), Bytes.toBytes("name"), Bytes.toBytes("hujinjun"));
//		put.add(Bytes.toBytes("author"), Bytes.toBytes("nickname"), Bytes.toBytes("一叶渡江"));
//		((Object) hAdmin).put(put);
		createTable2();
		
	}
	public static void createTable2()
	{
		String tableName = "TH10_2M";
		boolean dynamic = false;
		int recordsNumber = 2000000;
		ArrayList<String> columnFamilies = new ArrayList<String>();
		columnFamilies.add("cf1");
		columnFamilies.add("cf2");
		columnFamilies.add("cf3");
		columnFamilies.add("cf4");
		columnFamilies.add("cf5");
		columnFamilies.add("cf6");
		columnFamilies.add("cf7");
		columnFamilies.add("cf8");
		columnFamilies.add("cf9");
		columnFamilies.add("cf10");
		
		ArrayList<String> clms = new ArrayList<String>();
		clms.add("c1");
		clms.add("c2");
		clms.add("c3");
		clms.add("c4");
		clms.add("c5");
		clms.add("c6");
		clms.add("c7");
		clms.add("c8");
		clms.add("c9");
		clms.add("c10");
		
//		MetaDataHandler mdh = new MetaDataHandler();
		try {
			if(MetaDataHandler.isTableExistsOld(tableName))
			{
				MetaDataHandler.dropTable(tableName);
			}
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		try {
			MetaDataHandler.createTable(tableName, columnFamilies);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		insertRandomData1(tableName, columnFamilies, clms, recordsNumber, dynamic);
		
		//now begin to insert data
	}
	public static void insertRandomData1(String table_name, ArrayList<String> cfs, ArrayList<String> clms, int recordsNumber, boolean dynamic)
	{
		HTable table = null;
		try {
			table = new HTable(configuration,table_name);
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}

		Random rand = new Random();

		ArrayList<Put> puts = new ArrayList<Put>();
		for(int i=0; i < recordsNumber; i++)
		{
			String rowkey = new Integer(i+1).toString();
			Put put=new Put(rowkey.getBytes());
			
			for(int j=0; j < cfs.size(); j++)
			{
				if(dynamic)
				{
						int value = rand.nextInt(recordsNumber+1);
						put.add(cfs.get(j).getBytes(),clms.get(j).getBytes(),new Integer(value).toString().getBytes());
				}
				else
				{
					for(int k=0; k < clms.size(); k++)
					{
						int value = rand.nextInt(recordsNumber+1);
						put.add(cfs.get(j).getBytes(),clms.get(k).getBytes(),new Integer(value).toString().getBytes());
					}
					break;
				}
			}
			puts.add(put);

			if(i%10000==0)
			{
				try {
					table.put(puts);
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				puts.clear();
				System.out.println("Insert " + i);
			}
		}
		if(puts.size()>0)
		{
			try {
				table.put(puts);
				puts.clear();
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
		try {
			table.flushCommits();
		} catch (IOException e2) {
			// TODO Auto-generated catch block
			e2.printStackTrace();
		}
		table.clearRegionCache();
		try {
			table.close();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		try {
			table.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

}
